package com.mqstudy.kafka.consumer;

import com.mqstudy.common.model.Message;
import com.mqstudy.kafka.config.KafkaConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.util.concurrent.atomic.AtomicLong;

/**
 * Kafka消费者
 */
@Slf4j
@Component
public class KafkaConsumer {
    
    private final AtomicLong messageCount = new AtomicLong(0);
    
    /**
     * 测试主题消费者
     */
    @KafkaListener(topics = KafkaConfig.TEST_TOPIC, groupId = "test-group")
    public void handleTestMessage(@Payload Message message,
                                 @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                                 @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                                 @Header(KafkaHeaders.OFFSET) long offset,
                                 Acknowledgment ack) {
        try {
            log.info("收到测试消息 - Topic: {}, Partition: {}, Offset: {}, Message: {}", 
                    topic, partition, offset, message);
            
            processMessage(message);
            
            // 手动确认消息
            ack.acknowledge();
            log.info("测试消息处理完成，已确认");
            
        } catch (Exception e) {
            log.error("处理测试消息失败: {}", e.getMessage(), e);
            // 不确认消息，让它重新消费
        }
    }
    
    /**
     * 订单主题消费者
     */
    @KafkaListener(topics = KafkaConfig.ORDER_TOPIC, groupId = "order-group")
    public void handleOrderMessage(@Payload Message message,
                                  @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                                  @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                                  @Header(KafkaHeaders.OFFSET) long offset,
                                  Acknowledgment ack) {
        try {
            log.info("收到订单消息 - Topic: {}, Partition: {}, Offset: {}, Message: {}", 
                    topic, partition, offset, message);
            
            processOrderMessage(message);
            
            ack.acknowledge();
            log.info("订单消息处理完成，已确认");
            
        } catch (Exception e) {
            log.error("处理订单消息失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 用户主题消费者
     */
    @KafkaListener(topics = KafkaConfig.USER_TOPIC, groupId = "user-group")
    public void handleUserMessage(@Payload Message message,
                                 @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                                 @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                                 @Header(KafkaHeaders.OFFSET) long offset,
                                 Acknowledgment ack) {
        try {
            log.info("收到用户消息 - Topic: {}, Partition: {}, Offset: {}, Message: {}", 
                    topic, partition, offset, message);
            
            processUserMessage(message);
            
            ack.acknowledge();
            log.info("用户消息处理完成，已确认");
            
        } catch (Exception e) {
            log.error("处理用户消息失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 日志主题消费者
     */
    @KafkaListener(topics = KafkaConfig.LOG_TOPIC, groupId = "log-group")
    public void handleLogMessage(@Payload Message message,
                                @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                                @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                                @Header(KafkaHeaders.OFFSET) long offset,
                                Acknowledgment ack) {
        try {
            log.info("收到日志消息 - Topic: {}, Partition: {}, Offset: {}, Message: {}", 
                    topic, partition, offset, message);
            
            processLogMessage(message);
            
            ack.acknowledge();
            log.info("日志消息处理完成，已确认");
            
        } catch (Exception e) {
            log.error("处理日志消息失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 流式主题消费者
     */
    @KafkaListener(topics = KafkaConfig.STREAM_TOPIC, groupId = "stream-group")
    public void handleStreamMessage(@Payload Message message,
                                   @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                                   @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                                   @Header(KafkaHeaders.OFFSET) long offset,
                                   Acknowledgment ack) {
        try {
            log.info("收到流式消息 - Topic: {}, Partition: {}, Offset: {}, Message: {}", 
                    topic, partition, offset, message);
            
            processStreamMessage(message);
            
            ack.acknowledge();
            log.info("流式消息处理完成，已确认");
            
        } catch (Exception e) {
            log.error("处理流式消息失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 处理通用消息
     */
    private void processMessage(Message message) {
        long count = messageCount.incrementAndGet();
        log.info("开始处理消息: {}, 总处理数量: {}", message.getMessageId(), count);
        
        // 模拟业务处理时间
        try {
            Thread.sleep(50);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        log.info("消息处理完成: {}", message.getMessageId());
    }
    
    /**
     * 处理订单消息
     */
    private void processOrderMessage(Message message) {
        log.info("开始处理订单消息: {}", message.getMessageId());
        
        // 模拟订单处理逻辑
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        log.info("订单消息处理完成: {}", message.getMessageId());
    }
    
    /**
     * 处理用户消息
     */
    private void processUserMessage(Message message) {
        log.info("开始处理用户消息: {}", message.getMessageId());
        
        // 模拟用户处理逻辑
        try {
            Thread.sleep(80);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        log.info("用户消息处理完成: {}", message.getMessageId());
    }
    
    /**
     * 处理日志消息
     */
    private void processLogMessage(Message message) {
        log.info("开始处理日志消息: {}", message.getMessageId());
        
        // 模拟日志处理逻辑（快速处理）
        try {
            Thread.sleep(10);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        log.info("日志消息处理完成: {}", message.getMessageId());
    }
    
    /**
     * 处理流式消息
     */
    private void processStreamMessage(Message message) {
        log.info("开始处理流式消息: {}", message.getMessageId());
        
        // 模拟流式处理逻辑
        try {
            Thread.sleep(30);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        log.info("流式消息处理完成: {}", message.getMessageId());
    }
    
    /**
     * 获取消息处理统计
     */
    public long getMessageCount() {
        return messageCount.get();
    }
} 